package test

import java.util

import entity.Movie
import org.apache.commons.lang.StringUtils
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.{ArrayBuffer, ListBuffer}
import scala.collection.JavaConverters._

class selectMovies extends App {

  def selectAll(): java.util.List[Movie] = {
    val conf = new SparkConf().setAppName("MovieSerch").setMaster("local")
    //    val sc = new SparkContext(conf)
    val sc = SparkContext.getOrCreate(conf)
    //    val dataFile = "hdfs://localhost:9000/input/movies.csv"
    val dataFile = "file:///f:\\spark\\movie\\movies.csv"
    //过滤标题行
    var ML = sc.textFile(dataFile).filter(mv_1 => !mv_1.equals("movieId,title,genres")).collect()
    //通过“>”切分属性列
    var MD = ML.map(mv1 => mv1.replace(",", ">")).filter(temp => temp.split(">").length == 3)
    //转存到对象中
    var movieLIst: java.util.List[Movie] = null
    var bufferlist = ListBuffer[Movie]()
    for (movie <- MD) {
      var buffer: Movie = new Movie()
      //      var idIndex: Int = movie.indexOf(">", 0)
      var titleIndex: Int = movie.indexOf(">", 0)
      var TypeIndex: Int = movie.indexOf(">", titleIndex + 1)
      buffer.setMovieId(movie.substring(0, titleIndex))
      buffer.setTitle(movie.substring(titleIndex + 1, TypeIndex + titleIndex - 1))
      buffer.setGenres(movie.substring(TypeIndex + titleIndex))
      bufferlist += buffer
    }
    sc.stop()
    return bufferlist.toList.asJava
  }

  def selectOption(Type: String, Word: String): java.util.List[Movie] = {
    var movieList: java.util.List[Movie] = null
    val conf = new SparkConf().setAppName("MovieSerch").setMaster("local")
    //    val sc = new SparkContext(conf)
    val sc = SparkContext.getOrCreate(conf)
    val dataFile = "file:///f:\\spark\\movie\\movies.csv"
    val data = sc.textFile(dataFile)

    var res = data.filter(line => line.contains(Word)).collect().toList
    println(res)
    val list = new util.ArrayList[Movie]()
    for (x <- res) {
      val arr = x.split(",")
      val movieId = arr(0)
      val genres = arr(2)
      val title = StringUtils.remove(StringUtils.remove(x, arr(0) + ","), "," + arr(arr.length - 1))
      val result = new Movie(movieId, title, genres)
      list.add(result)
    }
    sc.stop()
    return list
  }

}



